package zhongche;

import com.alibaba.fastjson.JSONObject;
import com.zhongche.bean.*;
import com.zhongche.util.DateUtil;
import com.zhongche.util.RulesUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.apache.poi.hssf.record.DVALRecord;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;


public class DemoCheckJob {

    private static final String SPEED_KEY = "s0915691";

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        String ruleString = RulesUtil.paresRuleByFile("rules/viewInfo.txt");
        List<Nodes> nodesList = RulesUtil.parseListNodesByString(ruleString);
        Map<String, List<String>> edgesMapByTargetKey = RulesUtil.getEdgesMapByTargetKey(ruleString);
        String jsonStr = "{\n" +
                "  \"box_num\": 1,\n" +
                "  \"current_station\": 5,\n" +
                "  \"dDevice\": 50,\n" +
                "  \"dataParam\": {\n" +
                "\"s0000108\": 0,\n" +
                "\"s0915691\": 75,\n" +
                " \"s1999915\": 1,\n" +
                " \"s0847918\": 4\n" +
                "  },\n" +
                "  \"data_sDeviceNo\": 20,\n" +
                "  \"frame_num\": 5657,\n" +
                "  \"line_id\": 2,\n" +
                "  \"line_num\": \"2\",\n" +
                "  \"message_type\": 5025,\n" +
                "  \"netLink_flag\": 0,\n" +
                "  \"next_station\": 0,\n" +
                "  \"ocs_time\": \"210716153627\",\n" +
                "  \"sDevice\": 40,\n" +
                "  \"sDeviceTime\": \"210716153626\",   \n" +
                "  \"second_circle\": 1,\n" +
                "  \"speed\": 0.0,\n" +
                "  \"start_station\": 5,\n" +
                "  \"terminal_station\": 28,\n" +
                "  \"train_id\": 159160,\n" +
                "  \"train_name\": \"159160\",\n" +
                "  \"train_num\": 159160,\n" +
                "  \"train_type\": \"A52\",\n" +
                "  \"ttype_id\": 10\n" +
                "}\n";
        String jsonStr2 = "{\n" +
                "  \"box_num\": 1,\n" +
                "  \"current_station\": 5,\n" +
                "  \"dDevice\": 50,\n" +
                "  \"dataParam\": {\n" +
                "\"s0000108\": 0,\n" +
                "\"s0915691\": 0,\n" +
                " \"s1999915\": 1,\n" +
                " \"s0847918\": 4\n" +
                "  },\n" +
                "  \"data_sDeviceNo\": 20,\n" +
                "  \"frame_num\": 5657,\n" +
                "  \"line_id\": 2,\n" +
                "  \"line_num\": \"2\",\n" +
                "  \"message_type\": 5025,\n" +
                "  \"netLink_flag\": 0,\n" +
                "  \"next_station\": 0,\n" +
                "  \"ocs_time\": \"210716153627\",\n" +
                "  \"sDevice\": 40,\n" +
                "  \"sDeviceTime\": \"210716155040\",   \n" +
                "  \"second_circle\": 1,\n" +
                "  \"speed\": 0.0,\n" +
                "  \"start_station\": 5,\n" +
                "  \"terminal_station\": 28,\n" +
                "  \"train_id\": 159160,\n" +
                "  \"train_name\": \"159160\",\n" +
                "  \"train_num\": 159160,\n" +
                "  \"train_type\": \"A52\",\n" +
                "  \"ttype_id\": 10\n" +
                "}\n";

        String jsonStr3 = "{\n" +
                "  \"box_num\": 1,\n" +
                "  \"current_station\": 5,\n" +
                "  \"dDevice\": 50,\n" +
                "  \"dataParam\": {\n" +
                "\"s0000108\": 0,\n" +
                "\"s0915691\": 60,\n" +
                "    \"s1999915\": 1,\n" +
                " \"s0847918\": 5\n" +
                "  },\n" +
                "  \"data_sDeviceNo\": 20,\n" +
                "  \"frame_num\": 5657,\n" +
                "  \"line_id\": 2,\n" +
                "  \"line_num\": \"2\",\n" +
                "  \"message_type\": 5025,\n" +
                "  \"netLink_flag\": 0,\n" +
                "  \"next_station\": 0,\n" +
                "  \"ocs_time\": \"210716153627\",\n" +
                "  \"sDevice\": 40,\n" +
                "  \"sDeviceTime\": \"210716155626\",   \n" +
                "  \"second_circle\": 1,\n" +
                "  \"speed\": 0.0,\n" +
                "  \"start_station\": 5,\n" +
                "  \"terminal_station\": 28,\n" +
                "  \"train_id\": 159160,\n" +
                "  \"train_name\": \"159160\",\n" +
                "  \"train_num\": 159160,\n" +
                "  \"train_type\": \"A52\",\n" +
                "  \"ttype_id\": 10\n" +
                "}\n";

        String jsonStr4 = "{\n" +
                "  \"box_num\": 1,\n" +
                "  \"current_station\": 5,\n" +
                "  \"dDevice\": 50,\n" +
                "  \"dataParam\": {\n" +
                "\"s0000108\": 0,\n" +
                "\"s0915691\": 0,\n" +
                "    \"s1999915\": 1,\n" +
                " \"s0847918\": 5\n" +
                "  },\n" +
                "  \"data_sDeviceNo\": 20,\n" +
                "  \"frame_num\": 5657,\n" +
                "  \"line_id\": 2,\n" +
                "  \"line_num\": \"2\",\n" +
                "  \"message_type\": 5025,\n" +
                "  \"netLink_flag\": 0,\n" +
                "  \"next_station\": 0,\n" +
                "  \"ocs_time\": \"210716153627\",\n" +
                "  \"sDevice\": 40,\n" +
                "  \"sDeviceTime\": \"210716160626\",   \n" +
                "  \"second_circle\": 1,\n" +
                "  \"speed\": 0.0,\n" +
                "  \"start_station\": 5,\n" +
                "  \"terminal_station\": 28,\n" +
                "  \"train_id\": 159160,\n" +
                "  \"train_name\": \"159160\",\n" +
                "  \"train_num\": 159160,\n" +
                "  \"train_type\": \"A52\",\n" +
                "  \"ttype_id\": 10\n" +
                "}\n";

        String jsonStr5 = "{\n" +
                "  \"box_num\": 1,\n" +
                "  \"current_station\": 5,\n" +
                "  \"dDevice\": 50,\n" +
                "  \"dataParam\": {\n" +
                "\"s0000108\": 0,\n" +
                "\"s0915691\": 60,\n" +
                "    \"s1999915\": 1,\n" +
                " \"s0847918\": 6\n" +
                "  },\n" +
                "  \"data_sDeviceNo\": 20,\n" +
                "  \"frame_num\": 5657,\n" +
                "  \"line_id\": 2,\n" +
                "  \"line_num\": \"2\",\n" +
                "  \"message_type\": 5025,\n" +
                "  \"netLink_flag\": 0,\n" +
                "  \"next_station\": 0,\n" +
                "  \"ocs_time\": \"210716153627\",\n" +
                "  \"sDevice\": 40,\n" +
                "  \"sDeviceTime\": \"210716160626\",   \n" +
                "  \"second_circle\": 1,\n" +
                "  \"speed\": 0.0,\n" +
                "  \"start_station\": 5,\n" +
                "  \"terminal_station\": 28,\n" +
                "  \"train_id\": 159160,\n" +
                "  \"train_name\": \"159160\",\n" +
                "  \"train_num\": 159160,\n" +
                "  \"train_type\": \"A52\",\n" +
                "  \"ttype_id\": 10\n" +
                "}\n";
        String jsonStr6 = "{\n" +
                "  \"box_num\": 1,\n" +
                "  \"current_station\": 5,\n" +
                "  \"dDevice\": 50,\n" +
                "  \"dataParam\": {\n" +
                "\"s0000108\": 0,\n" +
                "\"s0915691\": 0,\n" +
                "    \"s1999915\": 1,\n" +
                " \"s0847918\": 6\n" +
                "  },\n" +
                "  \"data_sDeviceNo\": 20,\n" +
                "  \"frame_num\": 5657,\n" +
                "  \"line_id\": 2,\n" +
                "  \"line_num\": \"2\",\n" +
                "  \"message_type\": 5025,\n" +
                "  \"netLink_flag\": 0,\n" +
                "  \"next_station\": 0,\n" +
                "  \"ocs_time\": \"210716153627\",\n" +
                "  \"sDevice\": 40,\n" +
                "  \"sDeviceTime\": \"210716163626\",   \n" +
                "  \"second_circle\": 1,\n" +
                "  \"speed\": 0.0,\n" +
                "  \"start_station\": 5,\n" +
                "  \"terminal_station\": 28,\n" +
                "  \"train_id\": 159160,\n" +
                "  \"train_name\": \"159160\",\n" +
                "  \"train_num\": 159160,\n" +
                "  \"train_type\": \"A52\",\n" +
                "  \"ttype_id\": 10\n" +
                "}\n";

        DataStream<TrainInfo> sourceStream = getSourceStream(env, jsonStr, jsonStr2, jsonStr3,jsonStr4,jsonStr5,jsonStr6);

        tranProcessByRule(sourceStream.keyBy(TrainInfo::getTrainId),nodesList,edgesMapByTargetKey);

        env.execute();
    }


    public static DataStream<TrainInfo> getSourceStream(StreamExecutionEnvironment env, String... elements) {
        return env.fromElements(elements)
                .map(new MapFunction<String, TrainInfo>() {
                    @Override
                    public TrainInfo map(String value) throws Exception {
                        return JSONObject.parseObject(value, TrainInfo.class);
                    }
                })
                .name("转换对象");
    }


    public static void tranProcessByRule(DataStream<TrainInfo> dataStream, List<Nodes> nodeList,Map<String, List<String>> edgesMapByTargetKey) {
        Tuple2<List<FunctionCompare>, String> tuple2 = parseArgsByListNodes(nodeList, edgesMapByTargetKey);
        System.out.println(tuple2.f1);
        if(CalculateType.CONDITION_KEEP.equals(tuple2.f1)){
            List<FunctionCompare> list = tuple2.f0;
            String changeKey = "";
            String valueKey = "";
            Integer speed = 0;
            String compareType = "";
            if(list.size() == 2){
                boolean equals = list.get(0).equals(list.get(1));
                if(equals){
                    FunctionCompare functionCompare = list.get(0);
                    ArgumentsType argumentsType = functionCompare.getArguments().get(0);
                    if("digital".equals(argumentsType.getDataType())){
                        changeKey = argumentsType.getDataValue();
                    }
                    FunctionCompare lastFunctionCompare = functionCompare.getLastFunctionCompare();
                    if(lastFunctionCompare != null){
                        compareType = lastFunctionCompare.getCalculateName();
                        List<ArgumentsType> arguments = lastFunctionCompare.getArguments();
                        for (ArgumentsType arg: arguments) {
                            if("analog".equals(arg.getDataType())){
                                valueKey = arg.getDataValue();
                            }else if ("INT".equals(arg.getDataType())){
                                speed = Integer.parseInt(arg.getDataValue());
                            }
                        }
                    }
                }
            }
            dataStream.flatMap(new MyRichFloatMapFunction(changeKey,valueKey,speed,compareType))
                    .print();
        }


    }



    public static Tuple2<List<FunctionCompare>, String> parseArgsByListNodes(List<Nodes> nodeList, Map<String, List<String>> edgesMap) {
        Map<String, Nodes> nodesMap = RulesUtil.nodesListToNodesMap(nodeList);
        Map<String, String> edgesMapOver = edgesMapOver(edgesMap);
        String resourceNodeId = RulesUtil.getResourceNodeId(nodeList);
        List<String> list = edgesMap.get(resourceNodeId);
        List<FunctionCompare> functionCompareList = new ArrayList<>();
        String computeOperator = "";
        if (list.size() == 1) {
            Nodes nodes = nodesMap.get(list.get(0));
            computeOperator = nodes.getValue();
            List<Nodes> inputFunctionNodes = findInputFunctionNodes(nodesMap);
            Map<String, FunctionCompare> nodesArgsMap = new HashMap<>();
            for (Nodes node : inputFunctionNodes) {
                String value = node.getValue();
                List<DataTypeInfo> input = node.getInput();
                List<ArgumentsType> argsList = new ArrayList<>();
                for (DataTypeInfo info : input) {
                    Nodes nodes1 = nodesMap.get(info.getSourceId());
                    argsList.add(new ArgumentsType(nodes1.getDataType(), nodes1.getValue()));
                }
                FunctionCompare functionCompare = new FunctionCompare();
                functionCompare.setCalculateName(value);
                functionCompare.setArguments(argsList);
                nodesArgsMap.put(node.getId(), functionCompare);
            }
//            System.out.println(nodesArgsMap.toString());
            for (String key : nodesArgsMap.keySet()) {
                Nodes nodes1 = nodesMap.get(edgesMapOver.get(key));
                List<DataTypeInfo> input = nodes1.getInput();
                FunctionCompare functionCompare = new FunctionCompare();
                functionCompare.setCalculateName(nodes1.getValue());
                functionCompare.setLastFunctionCompare(nodesArgsMap.get(key));
                List<ArgumentsType> argsList = new ArrayList<>();
                for (DataTypeInfo info : input) {
                    if (!info.getSourceId().equals(key)) {
                        Nodes nodes2 = nodesMap.get(info.getSourceId());
                        if ("input".equals(nodes2.getShape())) {
                            argsList.add(new ArgumentsType(nodes2.getDataType(), nodes2.getValue()));
                        }
                    }
                }
                functionCompare.setArguments(argsList);
                functionCompareList.add(functionCompare);
            }
        }
        return Tuple2.of(functionCompareList, computeOperator);
    }


    /**
     * 获取input nodes
     *
     * @param dataTypeInfo
     * @param nodesMap
     * @return
     */
    public static List<Nodes> parseInput(DataTypeInfo dataTypeInfo, Map<String, Nodes> nodesMap) {
        Nodes nodes = nodesMap.get(dataTypeInfo.getSourceId());
        List<DataTypeInfo> input = nodes.getInput();
        List<Nodes> inputNodes = new ArrayList<>();
        while (input != null && input.size() > 0) {
            List<DataTypeInfo> newInfo = new ArrayList<>();
            for (DataTypeInfo info : input) {
                Nodes nodes1 = nodesMap.get(info.getSourceId());
                if ("input".equals(nodes1.getShape())) {
                    inputNodes.add(nodes1);
                }
                if (nodes1.getInput() != null) {
                    newInfo.addAll(nodes1.getInput());
                }

            }
            input = newInfo;

        }
        return inputNodes;

    }


    private static List<Nodes> findInputFunctionNodes(Map<String, Nodes> nodesMap) {
        List<Nodes> inputNodes = new ArrayList<>();
        for (Nodes nodes : nodesMap.values()) {
            if ("function".equals(nodes.getShape())) {
                List<DataTypeInfo> input = nodes.getInput();
                boolean flag = true;
                if (input != null && input.size() > 0) {
                    for (DataTypeInfo info : input) {
                        Nodes nodes1 = nodesMap.get(info.getSourceId());
                        if (!"input".equals(nodes1.getShape())) {
                            flag = false;
                        }
                    }
                    if (flag) {
                        inputNodes.add(nodes);
                    }
                }
            }
        }
        return inputNodes;
    }


    private static Map<String, String> edgesMapOver(Map<String, List<String>> edgesMap) {
        Map<String, String> edgesSourceKyMap = new HashMap<>();
        for (Map.Entry<String, List<String>> entry : edgesMap.entrySet()) {
            for (String str : entry.getValue()) {
                edgesSourceKyMap.put(str, entry.getKey());
            }

        }
        return edgesSourceKyMap;
    }


    public static class MyRichFloatMapFunction extends RichFlatMapFunction<TrainInfo, String> {


        private String changeKey;

        private Integer changeValue;

        private String compareType;


        private ValueState<Long> timeSate;

        private String valueKey;

        private Integer speed;

        public MyRichFloatMapFunction(String changeKey, String valueKey, Integer speed, String compareType) {
            this.changeKey = changeKey;
            this.valueKey = valueKey;
            this.speed = speed;
            this.compareType = compareType;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            timeSate = getRuntimeContext().getState(new ValueStateDescriptor<Long>("time-state", Long.class));
        }


        @Override
        public void flatMap(TrainInfo value, Collector<String> out) throws Exception {
            System.out.println(changeValue);
            Map<String, Integer> dataParam = value.getDataParam();
            Integer integer = dataParam.get(valueKey);
            Long aLong = DateUtil.transDateStrToStamp(value.getsDeviceTime());
            if ("eq".equals(compareType)) {
                if (integer != null && integer.compareTo(speed) == 0) {
                    if (timeSate.value() == null) {
                        timeSate.update(aLong);
                        this.changeValue = dataParam.get(this.changeKey);
                    } else {
                        if (!this.changeValue.equals(dataParam.get(this.changeKey))) {
                            this.changeValue = dataParam.get(this.changeKey);
                            Long intervalTime = (aLong - timeSate.value()) / 1000;
                            String record = value.getTrainName() + "(" + value.getTrainId() + "): 停车时间为" + intervalTime + "s";
                            out.collect(record);
                        }
                    }
                }
            }

        }
    }
}
